/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.example.hrpc.rpc.client;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.SettableFuture;
import io.netty.channel.Channel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import org.example.hrpc.rpc.core.Invocation;
import org.example.hrpc.rpc.message.MessageType;
import org.example.hrpc.rpc.message.Request;
import org.example.hrpc.rpc.message.SerializationType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.*;

import static org.example.hrpc.common.utils.NettyUtils.getRemoteAddress;


/**
 */
public class   TransportClient implements Closeable {
    private static final Logger logger = LoggerFactory.getLogger(TransportClient.class);

    private final Channel channel;
    @Nullable
    private String clientId;
    private volatile boolean timedOut;



    public TransportClient(Channel channel) {
        this.channel = Preconditions.checkNotNull(channel);
        this.timedOut = false;
    }

    public Channel getChannel() {
        return channel;
    }

    public boolean isActive() {
        return !timedOut && (channel.isOpen() || channel.isActive());
    }

    public SocketAddress getSocketAddress() {
        return channel.remoteAddress();
    }


    public String getClientId() {
        return clientId;
    }


    public void setClientId(String id) {
        Preconditions.checkState(clientId == null, "Client ID has already been set.");
        this.clientId = id;
    }



//    public long sendRpc(ByteBuffer message, RpcResponseCallback callback) {
//        if (logger.isTraceEnabled()) {
//            logger.trace("Sending RPC to {}", getRemoteAddress(channel));
//        }
//
//        long requestId = requestId();
//        // 添加requestID与RpcResponseCallback的引用之间的关系
//        // 这里的Handler是通过构造函数 传入的 那个 TransportResponseHandler
////    handler.addRpcRequest(requestId, callback);
////    RpcChannelListener listener = new RpcChannelListener(requestId, callback);
//        channel.writeAndFlush(new RpcRequest(requestId, new NioManagedBuffer(message)))
//                .addListener(listener);
//
//        return requestId;
//    }


    public CompletableFuture<?> sendRpc(Invocation invocation, int timeout, ExecutorService executor, RpcResponseCallback callback) {
        if (logger.isTraceEnabled()) {
            logger.trace("Sending RPC to {}", getRemoteAddress(channel));
        }

        long requestId = requestId();
        // 添加requestID与RpcResponseCallback的引用之间的关系
        // 这里的Handler是通过构造函数 传入的 那个 TransportResponseHandler
//    handler.addRpcRequest(requestId, callback);
//    RpcChannelListener listener = new RpcChannelListener(requestId, callback);
//        channel.writeAndFlush(new RpcRequest(requestId, new NioManagedBuffer(message)))
//                .addListener(listener);
        Request request = new Request();
        request.setRequestId(requestId);
        request.setMessageType(MessageType.SERVICE_REQUEST);
        request.setSerializationType(SerializationType.HESSIAN);
        request.setData(invocation);

        DefaultFuture future = DefaultFuture.newFuture(this.channel, request, timeout, executor);

        channel.writeAndFlush(request);

        return future;
    }




    /** Mark this channel as having timed out. */
    public void timeOut() {
        this.timedOut = true;
    }


    @Override
    public void close() {
        // close is a local operation and should finish with milliseconds; timeout just to be safe
        channel.close().awaitUninterruptibly(10, TimeUnit.SECONDS);
    }

    @Override
    public String toString() {
        return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
                .append("remoteAdress", channel.remoteAddress())
                .append("clientId", clientId)
                .append("isActive", isActive())
                .toString();
    }

    /**
     * 使用UUID生成请求主键
     * @return
     */
    private static long requestId() {
        return Math.abs(UUID.randomUUID().getLeastSignificantBits());
    }

//    private class StdChannelListener
//            implements GenericFutureListener<Future<? super Void>> {
//        final long startTime;
//        final Object requestId;
//
//        StdChannelListener(Object requestId) {
//            this.startTime = System.currentTimeMillis();
//            this.requestId = requestId;
//        }
//
//        @Override
//        public void operationComplete(Future<? super Void> future) throws Exception {
//            if (future.isSuccess()) {
//                if (logger.isTraceEnabled()) {
//                    long timeTaken = System.currentTimeMillis() - startTime;
//                    logger.trace("Sending request {} to {} took {} ms", requestId,
//                            getRemoteAddress(channel), timeTaken);
//                }
//            } else {
//                String errorMsg = String.format("Failed to send RPC %s to %s: %s", requestId,
//                        getRemoteAddress(channel), future.cause());
//                logger.error(errorMsg, future.cause());
//                channel.close();
//                try {
//                    handleFailure(errorMsg, future.cause());
//                } catch (Exception e) {
//                    logger.error("Uncaught exception in RPC response callback handler!", e);
//                }
//            }
//        }
//
//        void handleFailure(String errorMsg, Throwable cause) throws Exception {}
//    }
//
//    private class RpcChannelListener extends StdChannelListener {
//        final long rpcRequestId;
//        final RpcResponseCallback callback;
//
//        RpcChannelListener(long rpcRequestId, RpcResponseCallback callback) {
//            super("RPC " + rpcRequestId);
//            this.rpcRequestId = rpcRequestId;
//            this.callback = callback;
//        }
//
//        @Override
//        void handleFailure(String errorMsg, Throwable cause) {
//            handler.removeRpcRequest(rpcRequestId);
//            callback.onFailure(new IOException(errorMsg, cause));
//        }
//    }

}
